package org.maven.biz;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.shared.SharedCount;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.CloseableUtils;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;

import com.google.common.base.Preconditions;

/**
 *********************************************** 
 * Copyright (c) 2016 VIP.COM All Rights Reserved. FileName：com.vip.sce.utilits.zookeeper.CuratorUtils.java Created On:
 * 2016年7月15日 下午5:22:58 Description:
 * @author kevin02.liu
 * @version 1.0
 *********************************************** 
 */
public class CuratorService {
	private  final String UTF_8 = "UTF-8";
	private  Logger logger = Logger.getLogger(CuratorService.class);
	private  CuratorFramework client;

	public  void init(String serverLists, int baseSleepTimeMilliseconds, int maxRetries,int maxSlemepTimeMilliseconds, String namespace){
		init(serverLists, baseSleepTimeMilliseconds, maxRetries, maxSlemepTimeMilliseconds, namespace, null);
	}
			
	public  void init(String serverLists, int baseSleepTimeMilliseconds, int maxRetries,
			int maxSlemepTimeMilliseconds, String namespace, String digest) {
		try {
			Builder builder = CuratorFrameworkFactory
					.builder()
					.connectString(serverLists)
					.retryPolicy(
							new ExponentialBackoffRetry(baseSleepTimeMilliseconds, maxRetries,
									maxSlemepTimeMilliseconds)).namespace(namespace);
			if(StringUtils.isNotBlank(digest)){
				builder.authorization("digest", digest.getBytes());
				builder.aclProvider(new ACLProvider() {
                    public List<ACL> getDefaultAcl() {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }
                    public List<ACL> getAclForPath(final String path) {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }
                });
			}
			
			client = builder.build();
			client.start();
			client.blockUntilConnected();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public  boolean getDistributedAtomicInteger(String path, Integer value) {
		SharedCount count = new SharedCount(client, path, 0);
		try {
			boolean result = count.trySetCount(count.getVersionedValue(), value);
			return result;
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		} finally {
			try {
				count.close();
			} catch (IOException e) {
				logger.error(e.getMessage(), e);
			}
		}
		return false;
	}

	public  boolean getDistributedAtomicLong(String path, Long value) {
		DistributedAtomicLong count = new DistributedAtomicLong(client, path, new RetryNTimes(10, 10));
		try {
			AtomicValue<Long> result = count.add(value);
			if (result.succeeded()) {
				// Long preValue = result.preValue();
				// Long postValue = result.postValue();
				return result.succeeded();
			}
			return result.succeeded();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		return false;
	}

	public  void close() {
		if(null == client){
			return;
		}
		CloseableUtils.closeQuietly(client);
	}

	public String get(final String key) {
		return getDirectly(key);
	}

	public String getDirectly(final String key) {
		try {
			return new String(client.getData().forPath(key), Charset.forName(UTF_8));
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			return null;
		}
	}

	public  List<String> getChildrenKeys(final String key) {
		try {
			List<String> result = client.getChildren().forPath(key);
			Collections.sort(result, new Comparator<String>() {
				public int compare(final String o1, final String o2) {
					return o2.compareTo(o1);
				}
			});
			return result;
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			return Collections.emptyList();
		}
	}

	public  boolean isExisted(final String path) {
		try {
			return null != client.checkExists().forPath(path);
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
			return false;
		}
	}

	public  void persist(final String key, final String value) {
		try {
			if (!isExisted(key)) {
				create(key, value);
			} else {
				update(key, value);
			}
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
	}

	public  void create(final String key, final String value) throws Exception {
		client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(key, value.getBytes());
	}

	public  void update(final String key, final String value) {
		try {
			client.inTransaction().check().forPath(key).and().setData()
					.forPath(key, value.getBytes(Charset.forName(UTF_8))).and().commit();
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
	}

	public  void persistEphemeral(final String key, final String value) {
		try {
			if (isExisted(key)) {
				client.delete().deletingChildrenIfNeeded().forPath(key);
			}
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
					.forPath(key, value.getBytes(Charset.forName(UTF_8)));
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
	}

	public void persistEphemeralSequential(final String key) {
		try {
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(key);
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
	}

	public void remove(final String key) {
		try {
			client.delete().deletingChildrenIfNeeded().forPath(key);
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
	}

	public long getRegistryCenterTime(final String key) {
		long result = 0L;
		try {
			String path = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
					.forPath(key);
			result = client.checkExists().forPath(path).getCtime();
			// CHECKSTYLE:OFF
		} catch (final Exception ex) {
			// CHECKSTYLE:ON
		}
		Preconditions.checkState(0L != result, "Cannot get registry center time.");
		return result;
	}

	public  CuratorFramework getClient() {
		return client;
	}

	public  void setClient(CuratorFramework client) {
		this.client = client;
	}

	public  void NodeCache(String path) {
		try {
			final NodeCache nodeCache = new NodeCache(client, path, false);
			nodeCache.start();
			nodeCache.getListenable().addListener(new NodeCacheListener() {

				public void nodeChanged() throws Exception {
					logger.info(nodeCache.getCurrentData());
				}
			});
			nodeCache.close();
		} catch (Exception e) {
		}
	}

	public  void PathChildrenCache(String path) {
		try {
			PathChildrenCache cache = new PathChildrenCache(client, path, true);
			cache.getListenable().addListener(new PathChildrenCacheListener() {
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					switch (event.getType()) {
					case CHILD_ADDED:
						logger.info(event.getData().getPath());
						break;
					case CHILD_UPDATED:
						logger.info(event.getData().getPath());
						break;
					case CHILD_REMOVED:
						logger.info(event.getData().getPath());
						break;
					default:
						break;
					}
				}
			});
			cache.start(StartMode.POST_INITIALIZED_EVENT);
			cache.close();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public  void LeaderSelector(String path) {
		LeaderSelector leaderSelector = new LeaderSelector(client, path, new LeaderSelectorListener() {
			public void stateChanged(CuratorFramework arg0, ConnectionState arg1) {
				logger.info("stateChanged");
			}

			public void takeLeadership(CuratorFramework client) throws Exception {
				logger.info("takeLeadership");
			}
		});
		leaderSelector.autoRequeue();
		leaderSelector.start();
		leaderSelector.close();
	}

	 int l = 0;

	public  void DistributedLock(String path) {
		InterProcessMutex lock = new InterProcessMutex(client, path);
		try {
			lock.acquire();
			lock.release();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public  AtomicValue<Integer> distributedInteger(String path, Integer data) {
		DistributedAtomicInteger atomicInteger = new DistributedAtomicInteger(client, path, new RetryNTimes(3, 100));
		try {
			AtomicValue<Integer> result = atomicInteger.add(data);
//			int preValue = result.preValue();
//			int postValue = result.postValue();
			return result;
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
		return null;
	}

	public  void DistributedBarrier(String path, Integer data) {
		DistributedBarrier barrier = new DistributedBarrier(client, path);
		try {
			barrier.setBarrier();
			barrier.waitOnBarrier();// 等待barrier为0
			barrier.removeBarrier();// 清空计数器
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public  void DistributedDoubleBarrier(String path, Integer data) {
		DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, path, 5);
		try {
			barrier.enter();
			barrier.leave();// 清空计数器
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}
	
	public  void getAllPath(String servers, String namespace, String digest){
		try {
			init(servers, 100, 3, 300, namespace, digest);
			getJobDetail("/");
			close();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public  void getJobDetail(String path) throws Exception {
		CuratorFramework client = getClient();
		List<String> childrens  = client.getChildren().forPath(path);
		for(String children : childrens){
			String subPath = path +"/"+ children;
			if(!"/" .equals(path)){
				subPath = path +"/"+ children;
			}else{
				subPath = "/"+ children;
			}
			
			String value = new String(client.getData().forPath(subPath)).toString();
			logger.info(subPath + " | " + value);
			getJobDetail(subPath);
		}
	}


	public  void delete(String path) throws Exception {
		if(client.checkExists().forPath(path) != null){
			client.delete().deletingChildrenIfNeeded().forPath(path);
		}
	}

	public static void main(String[] args) {
		CuratorService service = new CuratorService();
		service.init("10.199.182.60:2181", 1000, 3, 10000, "live-lossLogic-job", "live-lossLogic-job");
		try {
			service.remove("/缺货工具源SQL分析");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
